package xyz.riceball.springcloudstreambinderfile.file;

import cn.hutool.core.io.FileUtil;
import org.springframework.cloud.stream.provisioning.ConsumerDestination;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import xyz.riceball.springcloudstreambinderfile.property.FileExtendedBindingProperties;

import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

// @checkstyle:off

/**
 * FileMessageProducer .
 *
 * @author xiaovcloud
 * @since 2024/6/13 14:09
 */
public class FileMessageProducerAdapter extends MessageProducerSupport {

	public static final String ARCHIVE = "archive.txt";
	private final ConsumerDestination destination;
	private final FileExtendedBindingProperties fileExtendedBindingProperties;
	ScheduledExecutorService executorService;
	private String previousPayload;

	public FileMessageProducerAdapter(ConsumerDestination destination, FileExtendedBindingProperties fileExtendedBindingProperties) {
		this.destination = destination;
		this.fileExtendedBindingProperties = fileExtendedBindingProperties;
	}

	@Override
	protected void onInit() {
		super.onInit();
		executorService = Executors.newScheduledThreadPool(1);
	}

	@Override
	public void doStart() {
		executorService.scheduleWithFixedDelay(() -> {
			String payload = getPayload();
			if (payload != null) {
				Message<String> receivedMessage = MessageBuilder.withPayload(payload).build();
				sendMessage(receivedMessage);
			}

		}, 0, 50, TimeUnit.MILLISECONDS);
	}

	@Override
	protected void doStop() {
		executorService.shutdownNow();
	}

	private String getPayload() {
		try {
			List<String> allLines = Files.readAllLines(Paths.get(fileExtendedBindingProperties.getPath() + File.separator + destination.getName() + ".txt"));
			String currentPayload = allLines.get(allLines.size() - 1);
			if (!currentPayload.equals(previousPayload)) {
				previousPayload = currentPayload;
				return currentPayload;
			}
		} catch (IOException e) {
			FileUtil.touch(new File(fileExtendedBindingProperties.getPath() + File.separator + destination.getName() + ".txt"));
		}

		return null;
	}
}
// @checkstyle:on
